package org.smartboot.sosa.rpc.netty;

import java.util.List;

import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.smartboot.sosa.core.util.StringUtils;
import org.smartboot.sosa.rpc.RpcConstant;
import org.smartboot.sosa.rpc.serial.SerializableBean;

import io.netty.buffer.ByteBuf;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.ByteToMessageDecoder;
import io.netty.handler.codec.DecoderException;
import io.netty.util.AttributeKey;
import io.protostuff.ProtobufIOUtil;
import io.protostuff.Schema;
import io.protostuff.runtime.RuntimeSchema;

/**
 * RPC消息解码
 * 
 * @author Seer
 * @version RequestDecoder.java, v 0.1 2015年11月20日 下午8:56:09 Seer Exp.
 */
public class RpcDecoder extends ByteToMessageDecoder {
	private static final String SERIALIZABLE_BEAN_KEY = "serilBean";
	private static final Logger LOGGER = LogManager.getLogger(RpcEncoder.class);

	@Override
	protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
		if (in.readableBytes() < RpcConstant.HEAD_LENGTH) {
			return;
		}
		// 校验幻数
		for (int i = 0; i < RpcConstant.MAGIC_NUM.length; i++) {
			if (in.getByte(i) != RpcConstant.MAGIC_NUM[i]) {
				byte[] b = new byte[in.readableBytes()];
				in.readBytes(b);
				System.err.println(StringUtils.toHexString(b));
				throw new DecoderException("invalid magic number");
			}
		}
		// 获取消息体大小
		int size = in.getInt(RpcConstant.MAGIC_NUM.length);
		if (size <= 0) {
			throw new DecoderException("invalid message size " + size);
		}

		// 消息体读取完毕再进行解码
		if (in.readableBytes() < size + RpcConstant.HEAD_LENGTH) {
			return;
		}

		byte[] bytes = new byte[size];
		in.skipBytes(RpcConstant.HEAD_LENGTH).readBytes(bytes);
		if (LOGGER.isDebugEnabled()) {
			LOGGER.debug("decode bit stream " + StringUtils.toHexString(bytes));
		}
		AttributeKey<SerializableBean> key = AttributeKey.valueOf(SERIALIZABLE_BEAN_KEY);
		Channel channel = ctx.channel();
		SerializableBean bean = channel.attr(key).get();
		if (bean == null) {
			bean = new SerializableBean();
			channel.attr(key).set(bean);
		}
		Schema<SerializableBean> schema = RuntimeSchema.getSchema(SerializableBean.class);
		ProtobufIOUtil.mergeFrom(bytes, bean, schema);
		out.add(bean.getBean());
		in.discardReadBytes();
	}

}
